//! 并发控制增强测试
//! 
//! 本模块测试事务系统在高并发场景下的正确性和性能

use super::transaction::*;
use super::page::PageManager;
use super::tuple::RowId;
use std::sync::{Arc, Barrier};
use std::thread;
use std::time::{Duration, Instant};
use tempfile::tempdir;

/// 并发控制测试套件
pub struct ConcurrencyTestSuite {
    transaction_manager: Arc<TransactionManager>,
    temp_dir: tempfile::TempDir,
}

impl ConcurrencyTestSuite {
    /// 创建新的并发测试套件
    pub fn new() -> Result<Self, crate::Error> {
        let temp_dir = tempdir().map_err(|e| crate::Error::storage(format!("创建临时目录失败: {}", e)))?;
        let wal_path = temp_dir.path().join("concurrency_test.wal").to_string_lossy().to_string();
        let page_manager = Arc::new(PageManager::new(2000));
        
        let transaction_manager = Arc::new(TransactionManager::new(page_manager, wal_path)?);
        
        Ok(Self {
            transaction_manager,
            temp_dir,
        })
    }
    
    /// 测试高并发事务处理
    pub fn test_high_concurrency_transactions(&self) -> Result<(), crate::Error> {
        println!("🧪 测试高并发事务处理...");
        
        let num_threads = 8;
        let transactions_per_thread = 100;
        let barrier = Arc::new(Barrier::new(num_threads));
        let tx_manager = self.transaction_manager.clone();
        
        let start_time = Instant::now();
        
        let handles: Vec<_> = (0..num_threads).map(|thread_id| {
            let tx_manager = tx_manager.clone();
            let barrier = barrier.clone();
            
            thread::spawn(move || -> Result<(), crate::Error> {
                // 等待所有线程准备就绪
                barrier.wait();
                
                for i in 0..transactions_per_thread {
                    let tx = tx_manager.begin_transaction(IsolationLevel::ReadCommitted, false)?;
                    
                    // 模拟一些工作
                    let mut write_set = tx.write_set.write();
                    write_set.insert(RowId::new(1, (thread_id * 1000 + i) as u16));
                    drop(write_set);
                    
                    // 随机决定提交或回滚
                    if i % 10 == 0 {
                        tx_manager.rollback_transaction(tx)?;
                    } else {
                        tx_manager.commit_transaction(tx)?;
                    }
                }
                
                Ok(())
            })
        }).collect();
        
        // 等待所有线程完成
        for handle in handles {
            handle.join().unwrap()?;
        }
        
        let elapsed = start_time.elapsed();
        let total_transactions = num_threads * transactions_per_thread;
        let throughput = total_transactions as f64 / elapsed.as_secs_f64();
        
        println!("  并发事务吞吐量: {:.2} 事务/秒", throughput);
        println!("  总事务数: {}", total_transactions);
        println!("  并发线程数: {}", num_threads);
        
        // 验证统计信息
        let stats = self.transaction_manager.get_stats();
        assert_eq!(stats.total_transactions, total_transactions as u64);
        
        println!("✅ 高并发事务测试通过");
        Ok(())
    }
    
    /// 测试死锁检测和解决
    pub fn test_deadlock_detection_and_resolution(&self) -> Result<(), crate::Error> {
        println!("🧪 测试死锁检测和解决...");
        
        let tx_manager = self.transaction_manager.clone();
        let barrier = Arc::new(Barrier::new(2));
        
        let tx_manager1 = tx_manager.clone();
        let tx_manager2 = tx_manager.clone();
        let barrier1 = barrier.clone();
        let barrier2 = barrier.clone();
        
        let handle1 = thread::spawn(move || -> Result<(), crate::Error> {
            let tx = tx_manager1.begin_transaction(IsolationLevel::ReadCommitted, false)?;
            
            // 获取锁A
            tx_manager1.get_lock_manager().acquire_lock(tx.id, "resource_A".to_string(), LockMode::Exclusive)?;
            
            // 等待另一个线程获取锁B
            barrier1.wait();
            thread::sleep(Duration::from_millis(10));
            
            // 尝试获取锁B（可能导致死锁）
            let result = tx_manager1.get_lock_manager().acquire_lock(tx.id, "resource_B".to_string(), LockMode::Exclusive);
            
            // 清理
            tx_manager1.get_lock_manager().release_lock(tx.id, &"resource_A".to_string())?;
            if result.is_ok() {
                tx_manager1.get_lock_manager().release_lock(tx.id, &"resource_B".to_string())?;
            }
            
            tx_manager1.rollback_transaction(tx)?;
            Ok(())
        });
        
        let handle2 = thread::spawn(move || -> Result<(), crate::Error> {
            let tx = tx_manager2.begin_transaction(IsolationLevel::ReadCommitted, false)?;
            
            // 获取锁B
            tx_manager2.get_lock_manager().acquire_lock(tx.id, "resource_B".to_string(), LockMode::Exclusive)?;
            
            // 等待另一个线程获取锁A
            barrier2.wait();
            thread::sleep(Duration::from_millis(10));
            
            // 尝试获取锁A（可能导致死锁）
            let result = tx_manager2.get_lock_manager().acquire_lock(tx.id, "resource_A".to_string(), LockMode::Exclusive);
            
            // 清理
            tx_manager2.get_lock_manager().release_lock(tx.id, &"resource_B".to_string())?;
            if result.is_ok() {
                tx_manager2.get_lock_manager().release_lock(tx.id, &"resource_A".to_string())?;
            }
            
            tx_manager2.rollback_transaction(tx)?;
            Ok(())
        });
        
        // 等待线程完成
        let result1 = handle1.join().unwrap();
        let result2 = handle2.join().unwrap();
        
        // 至少有一个线程应该成功（死锁被解决）
        assert!(result1.is_ok() || result2.is_ok());
        
        println!("✅ 死锁检测和解决测试通过");
        Ok(())
    }
    
    /// 测试读写并发性能
    pub fn test_read_write_concurrency(&self) -> Result<(), crate::Error> {
        println!("🧪 测试读写并发性能...");
        
        let num_readers = 4;
        let num_writers = 2;
        let operations_per_thread = 50;
        let barrier = Arc::new(Barrier::new(num_readers + num_writers));
        let tx_manager = self.transaction_manager.clone();
        
        let start_time = Instant::now();
        
        // 启动读线程
        let reader_handles: Vec<_> = (0..num_readers).map(|reader_id| {
            let tx_manager = tx_manager.clone();
            let barrier = barrier.clone();
            
            thread::spawn(move || -> Result<(), crate::Error> {
                barrier.wait();
                
                for i in 0..operations_per_thread {
                    let tx = tx_manager.begin_transaction(IsolationLevel::ReadCommitted, true)?;
                    
                    // 模拟读操作
                    let mut read_set = tx.read_set.write();
                    read_set.insert(RowId::new((reader_id * 100 + i) as u64, 0));
                    drop(read_set);
                    
                    tx_manager.commit_transaction(tx)?;
                }
                
                Ok(())
            })
        }).collect();
        
        // 启动写线程
        let writer_handles: Vec<_> = (0..num_writers).map(|writer_id| {
            let tx_manager = tx_manager.clone();
            let barrier = barrier.clone();
            
            thread::spawn(move || -> Result<(), crate::Error> {
                barrier.wait();
                
                for i in 0..operations_per_thread {
                    let tx = tx_manager.begin_transaction(IsolationLevel::ReadCommitted, false)?;
                    
                    // 模拟写操作
                    let mut write_set = tx.write_set.write();
                    write_set.insert(RowId::new((writer_id * 100 + i + 10000) as u64, 0));
                    drop(write_set);
                    
                    tx_manager.commit_transaction(tx)?;
                }
                
                Ok(())
            })
        }).collect();
        
        // 等待所有线程完成
        for handle in reader_handles {
            handle.join().unwrap()?;
        }
        for handle in writer_handles {
            handle.join().unwrap()?;
        }
        
        let elapsed = start_time.elapsed();
        let total_operations = (num_readers + num_writers) * operations_per_thread;
        let throughput = total_operations as f64 / elapsed.as_secs_f64();
        
        println!("  读写并发吞吐量: {:.2} 操作/秒", throughput);
        println!("  读线程数: {}, 写线程数: {}", num_readers, num_writers);
        
        println!("✅ 读写并发测试通过");
        Ok(())
    }
    
    /// 测试长事务和短事务混合场景
    pub fn test_mixed_transaction_lengths(&self) -> Result<(), crate::Error> {
        println!("🧪 测试长事务和短事务混合场景...");
        
        let tx_manager = self.transaction_manager.clone();
        let barrier = Arc::new(Barrier::new(3));
        
        // 长事务
        let long_tx_handle = {
            let tx_manager = tx_manager.clone();
            let barrier = barrier.clone();
            
            thread::spawn(move || -> Result<(), crate::Error> {
                barrier.wait();
                
                let tx = tx_manager.begin_transaction(IsolationLevel::ReadCommitted, false)?;
                
                // 模拟长时间运行的事务
                for i in 0..10 {
                    let mut write_set = tx.write_set.write();
                    write_set.insert(RowId::new(i + 20000, 0));
                    drop(write_set);
                    
                    thread::sleep(Duration::from_millis(10));
                }
                
                tx_manager.commit_transaction(tx)?;
                Ok(())
            })
        };
        
        // 短事务1
        let short_tx_handle1 = {
            let tx_manager = tx_manager.clone();
            let barrier = barrier.clone();
            
            thread::spawn(move || -> Result<(), crate::Error> {
                barrier.wait();
                
                for i in 0..20 {
                    let tx = tx_manager.begin_transaction(IsolationLevel::ReadCommitted, false)?;
                    
                    let mut write_set = tx.write_set.write();
                    write_set.insert(RowId::new(i + 30000, 0));
                    drop(write_set);
                    
                    tx_manager.commit_transaction(tx)?;
                    
                    thread::sleep(Duration::from_millis(1));
                }
                
                Ok(())
            })
        };
        
        // 短事务2
        let short_tx_handle2 = {
            let tx_manager = tx_manager.clone();
            let barrier = barrier.clone();
            
            thread::spawn(move || -> Result<(), crate::Error> {
                barrier.wait();
                
                for i in 0..20 {
                    let tx = tx_manager.begin_transaction(IsolationLevel::ReadCommitted, false)?;
                    
                    let mut write_set = tx.write_set.write();
                    write_set.insert(RowId::new(i + 40000, 0));
                    drop(write_set);
                    
                    tx_manager.commit_transaction(tx)?;
                    
                    thread::sleep(Duration::from_millis(1));
                }
                
                Ok(())
            })
        };
        
        // 等待所有线程完成
        long_tx_handle.join().unwrap()?;
        short_tx_handle1.join().unwrap()?;
        short_tx_handle2.join().unwrap()?;
        
        // 验证所有事务都成功完成
        let stats = self.transaction_manager.get_stats();
        assert!(stats.committed_transactions >= 41); // 1个长事务 + 40个短事务
        
        println!("✅ 混合事务长度测试通过");
        Ok(())
    }
    
    /// 运行完整的并发控制测试套件
    pub fn run_full_concurrency_test_suite(&self) -> Result<(), crate::Error> {
        println!("🚀 开始并发控制测试套件...");
        println!("=====================================");
        
        self.test_high_concurrency_transactions()?;
        self.test_deadlock_detection_and_resolution()?;
        self.test_read_write_concurrency()?;
        self.test_mixed_transaction_lengths()?;
        
        println!("=====================================");
        println!("🎉 并发控制测试套件全部通过！");
        
        // 输出最终统计信息
        let stats = self.transaction_manager.get_stats();
        println!("📊 最终统计:");
        println!("  总事务数: {}", stats.total_transactions);
        println!("  已提交事务: {}", stats.committed_transactions);
        println!("  已回滚事务: {}", stats.rolled_back_transactions);
        println!("  活跃事务: {}", stats.active_transactions);
        
        Ok(())
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    
    #[test]
    fn test_high_concurrency() {
        let test_suite = ConcurrencyTestSuite::new().unwrap();
        test_suite.test_high_concurrency_transactions().unwrap();
    }
    
    #[test]
    fn test_deadlock_detection() {
        let test_suite = ConcurrencyTestSuite::new().unwrap();
        test_suite.test_deadlock_detection_and_resolution().unwrap();
    }
    
    #[test]
    fn test_read_write_concurrency() {
        let test_suite = ConcurrencyTestSuite::new().unwrap();
        test_suite.test_read_write_concurrency().unwrap();
    }
    
    #[test]
    fn test_mixed_transaction_lengths() {
        let test_suite = ConcurrencyTestSuite::new().unwrap();
        test_suite.test_mixed_transaction_lengths().unwrap();
    }
    
    #[test]
    fn test_full_concurrency_suite() {
        let test_suite = ConcurrencyTestSuite::new().unwrap();
        test_suite.run_full_concurrency_test_suite().unwrap();
    }
}
